package org.mq.mymq.broker.core;

import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.model.MyMqTopicModel;
import org.mq.mymq.broker.model.QueueModel;
import org.mq.mymq.broker.utils.LogFileNameUtil;
import org.mq.mymq.broker.utils.PutMessageLock;
import org.mq.mymq.broker.utils.UnfairReentrantLock;
import org.mq.mymq.common.constants.BrokerConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName ConsumerQueueMMapFileModel
 * @Description 对consumeQueue文件做mmap映射的核心对象
 * @Author jiarong_ye
 * @Date 2025/6/16 20:11
 */
public class ConsumeQueueMMapFileModel {
    private static final Logger log = LoggerFactory.getLogger(ConsumeQueueMMapFileModel.class);
    private File file;
    private MappedByteBuffer mappedByteBuffer;
    private ByteBuffer readBuffer;
    private FileChannel fileChannel;
    private String topic;
    private Integer queueId;
    private String consumeQueueFileName;
    private PutMessageLock putMessageLock;


    public void loadFileInMMap(String topicName, int queueId, int startOffset, int latestWriteOffset, int mappedSize) throws IOException {
        this.topic = topicName;
        this.queueId = queueId;
        String filePath = getLatestConsumeQueueFile();
        this.doMMap(filePath, startOffset, latestWriteOffset, mappedSize);
        putMessageLock = new UnfairReentrantLock();
    }

    public void writeContent(byte[] content, boolean force) throws IOException {
        try {
            putMessageLock.lock();
            this.getLatestConsumeQueueFile();
            mappedByteBuffer.put(content);
            if (force) {
                mappedByteBuffer.force();
            }
        } finally {
            putMessageLock.unlock();
        }
    }

    public void writeContent(byte[] content) throws IOException {
        writeContent(content, false);
    }

    /**
     * 读取consumequeue数据内容
     *
     * @param pos 消息读取开始位置，消息在队列中存储的值为固定12字节的索引数据
     * @return
     */
    public byte[] readContent(int pos) {
        byte[] content = new byte[BrokerConstants.CONSUME_QUEUE_EACH_MSG_SIZE];
        ByteBuffer readBuf = readBuffer.slice();
        readBuf.position(pos);
        readBuf.get(content);
        return content;
    }

    /**
     * 读取consumequeue数据内容
     *
     * @param pos 消息读取开始位置，消息在队列中存储的值为固定12字节的索引数据
     * @param msgCount 读取消息条数
     * @return
     */
    public List<byte[]> readContent(int pos, int msgCount) {
        ByteBuffer readBuf = readBuffer.slice();
        readBuf.position(pos);
        List<byte[]> loadContentList = new ArrayList<>();
        for (int i = 0; i < msgCount; i++) {
            byte[] content = new byte[BrokerConstants.CONSUME_QUEUE_EACH_MSG_SIZE];
            readBuf.get(content);
            loadContentList.add(content);
        }
        return loadContentList;
    }


    /**
     * 执行mmap步骤
     *
     * @param filePath 待映射文件路径
     * @param startOffset 开始映射的offset
     * @param mappedSize 映射文件大小 单位：byte
     * @throws IOException
     */
    private void doMMap(String filePath, int startOffset, int latestWriteOffset, int mappedSize) throws IOException {
        file = new File(filePath);
        if (!file.exists() || file.isDirectory()) {
            throw new FileNotFoundException("filePath is " + filePath + " inValid");
        }
        this.fileChannel = new RandomAccessFile(file, "rw").getChannel();
        this.mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, startOffset, mappedSize);
        this.readBuffer = mappedByteBuffer.slice();
        this.mappedByteBuffer.position(latestWriteOffset);
    }

    /**
     * 获取最新commitLog文件
     *
     * @return
     * @throws IOException
     */
    private String getLatestConsumeQueueFile() throws IOException {
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(topic);
        if (myMqTopicModel == null) {
            throw new IllegalArgumentException("topicName is " + topic + " inValid");
        }
        QueueModel queueModel = myMqTopicModel.getQueueList().get(queueId);
        if (queueModel == null) {
            throw new IllegalArgumentException("queueId is " + queueId + " inValid");
        }
        int diff = queueModel.countDiff();
        String filePath = "";
        if (diff < 12) {
            // 当前文件已经写满
            filePath = this.createNewConsumeQueueFile(queueModel.getFileName());
            this.doMMap(filePath, 0, 0, queueModel.getOffsetLimit());
            queueModel.setFileName(LogFileNameUtil.incrConsumeQueueFileName(queueModel.getFileName()));
            queueModel.setLastOffset(0);
            queueModel.setLatestOffset(new AtomicInteger(0));
        } else if (diff >= 12) {
            filePath = LogFileNameUtil.buildConsumeQueueFilePath(topic, queueId, queueModel.getFileName());
        }
        return filePath;
    }

    private String createNewConsumeQueueFile(String consumeQueueFileName) throws IOException {
        String newFileName = LogFileNameUtil.incrConsumeQueueFileName(consumeQueueFileName);
        String newFilePath = LogFileNameUtil.buildConsumeQueueFilePath(topic, queueId, newFileName);
        File newCommitLogFile = new File(newFilePath);
        try {
            newCommitLogFile.createNewFile();
            log.info("创建的新的consumeQueue文件：{}", newFilePath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return newFilePath;
    }

    public void clean() {
        if (mappedByteBuffer == null || !mappedByteBuffer.isDirect() || mappedByteBuffer.capacity() == 0)
            return;
        invoke(invoke(viewed(mappedByteBuffer), "cleaner"), "clean");
    }
    private Object invoke(final Object target, final String methodName, final Class<?>... args) {
        return AccessController.doPrivileged(new PrivilegedAction<Object>() {
            public Object run() {
                try {
                    Method method = method(target, methodName, args);
                    method.setAccessible(true);
                    return method.invoke(target);
                } catch (Exception e) {
                    throw new IllegalStateException(e);
                }
            }
        });
    }

    private Method method(Object target, String methodName, Class<?>[] args)
            throws NoSuchMethodException {
        try {
            return target.getClass().getMethod(methodName, args);
        } catch (NoSuchMethodException e) {
            return target.getClass().getDeclaredMethod(methodName, args);
        }
    }

    private ByteBuffer viewed(ByteBuffer buffer) {
        String methodName = "viewedBuffer";
        Method[] methods = buffer.getClass().getMethods();
        for (int i = 0; i < methods.length; i++) {
            if (methods[i].getName().equals("attachment")) {
                methodName = "attachment";
                break;
            }
        }

        ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
        if (viewedBuffer == null)
            return buffer;
        else
            return viewed(viewedBuffer);
    }
    public File getFile() {
        return file;
    }

    public void setFile(File file) {
        this.file = file;
    }

    public MappedByteBuffer getMappedByteBuffer() {
        return mappedByteBuffer;
    }

    public void setMappedByteBuffer(MappedByteBuffer mappedByteBuffer) {
        this.mappedByteBuffer = mappedByteBuffer;
    }

    public FileChannel getFileChannel() {
        return fileChannel;
    }

    public void setFileChannel(FileChannel fileChannel) {
        this.fileChannel = fileChannel;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public Integer getQueueId() {
        return queueId;
    }

    public void setQueueId(Integer queueId) {
        this.queueId = queueId;
    }

    public String getConsumeQueueFileName() {
        return consumeQueueFileName;
    }

    public void setConsumeQueueFileName(String consumeQueueFileName) {
        this.consumeQueueFileName = consumeQueueFileName;
    }

    public PutMessageLock getPutMessageLock() {
        return putMessageLock;
    }

    public void setPutMessageLock(PutMessageLock putMessageLock) {
        this.putMessageLock = putMessageLock;
    }
}
